package com.nyist.mq.service.impl;

import com.nyist.entity.User;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.stereotype.Service;

/**
 * 1. 处理生产者本地事务信息
 *
 * @ClassName ProducerTransactionListenerImpl
 * @Description
 * @Author xiaoyang
 * @Date 2023/03/29/16:16
 */
@Slf4j
@Service
@RequiredArgsConstructor
@RocketMQTransactionListener // 默认会获取生产者组提交的事务信息
public class ProducerTransactionListenerImpl implements RocketMQLocalTransactionListener {

    private static Boolean isTransactionalSuccess = false;

    /**
     * 执行本地事务
     * @author yangxiaogen
     * @date 2023/3/29 16:32
     * @param msg 发送到消费者的信息
     * @param arg 本地信息
     * @return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        MessageHeaders messageHeaders = msg.getHeaders();
        String transactionId = (String) messageHeaders.get(RocketMQHeaders.TRANSACTION_ID);
        log.info("======>>>【执行本地事务】消息体参数：transactionId={}", transactionId);
        // 执行带有事务注解的本地方法：增加用户余额+保存mq日志
        try {
            User user = (User) arg;
            String assertUserId = "123";
            // userService.addBalance(userCharge, transactionId);
            String userId = user.getUserId();
            if (StringUtils.equals(assertUserId, userId)) {
                log.info("======>>>【执行本地事务】成功,userId={}", userId);
                isTransactionalSuccess = true;
            } else {
                isTransactionalSuccess = false;
                throw new IllegalArgumentException("执行本地事务失败,userId="+userId);
            }
            return RocketMQLocalTransactionState.COMMIT; // 正常：向MQ Server发送commit消息
        } catch (Exception e) {
            log.error("======>>>错误信息为:",e);
            log.error("【执行本地事务】发生异常，消息将被回滚,{}", e.getMessage());
            return RocketMQLocalTransactionState.ROLLBACK; // 异常：向MQ Server发送rollback消息
        }
    }

    /**
     * 用于回查本地事务执行结果
     * @author yangxiaogen
     * @date 2023/3/29 16:35
     * @param msg
     * @return org.apache.rocketmq.spring.core.RocketMQLocalTransactionState
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        MessageHeaders headers = msg.getHeaders();
        String transactionId = headers.get(RocketMQHeaders.TRANSACTION_ID, String.class);
        log.info("======>>>【回查本地事务】transactionId={}", transactionId);

        // 根据事务id查询事务日志表
        // MQTransactionLog mqTransactionLog = mqTransactionLogMapper.selectByPrimaryKey(transactionId);
        if (isTransactionalSuccess) { // 查到表明本地事务执行成功，提交
            return RocketMQLocalTransactionState.COMMIT;
        }
        return RocketMQLocalTransactionState.ROLLBACK; // 没查到表明本地事务执行失败,通知回滚
    }
}
